package cn.com.nes.common.util;


import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.pool.KryoPool;
import com.esotericsoftware.kryo.io.Input;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;

import java.util.List;

public class KryoDecoder extends MessageToMessageDecoder<ByteBuf> {

    private final KryoPool kryoPool;

    public KryoDecoder(KryoPool kryoPool) {
        if (kryoPool == null) {
            throw new NullPointerException("kryoPool");
        }
        this.kryoPool = kryoPool;
    }

    public KryoDecoder() {
        this(KryoUtil.newKryoPool());
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
        final byte[] array;
        final int offset;
        final int length = msg.readableBytes();
        if (msg.hasArray()) {
            array = msg.array();
            offset = msg.arrayOffset() + msg.readerIndex();
        } else {
            array = ByteBufUtil.getBytes(msg, msg.readerIndex(), length, false);
            offset = 0;
        }
        Kryo kryo = null;
        Input input = null;
        Object obj;
        try {
            input = new Input(array, offset, length);
            kryo = this.kryoPool.borrow();
            obj = kryo.readClassAndObject(input);
        } finally {
            if(input != null) {
                input.close();
            }
            if(kryo != null) {
                this.kryoPool.release(kryo);
            }
        }
        out.add(obj);
    }


}
